package flink_p1

import org.apache.flink.api.common.functions.ReduceFunction
import org.apache.flink.api.scala.createTypeInformation
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala.function.WindowFunction
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector

/**
 * watermark
 *
 * */

object FlinkTest_20_window_watermark {

  def main(args: Array[String]): Unit = {


    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    //指定使用EventTime
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    //设置插入watermark的时间间隔
    env.getConfig.setAutoWatermarkInterval(100)



    //可以自己实现
    // val watermarks = new AssignerWithPeriodicWatermarks[String]() {
    //
    //      val delay:Long = 3000L
    //      var maxTime:Long = _
    //
    //      override def getCurrentWatermark: Watermark = {
    //        new Watermark(maxTime - delay)
    //      }
    //
    //      override def extractTimestamp(element: String, previousElementTimestamp: Long): Long = {
    //        val time: String = element.split(" ")(0)
    //        maxTime = maxTime.max(time.toLong)
    //        time.toLong
    //      }
    //    }


    //也可以用现成的实现类
    val watermarks = new BoundedOutOfOrdernessTimestampExtractor[String](Time.seconds(3)) {
      override def extractTimestamp(element: String): Long = {
        val time: String = element.split(" ")(0)
        time.toLong
      }
    }


    val socketStream: DataStream[String] = env.socketTextStream("127.0.0.1", 8889)
      .assignTimestampsAndWatermarks(watermarks)

    //assignTimestampsAndWatermarks 指定使用watermark，

    socketStream
      .map((_, 1))
      .keyBy(_._1)
      .timeWindow(Time.seconds(3))
      .reduce(new ReduceFunction[(String, Int)] {
        override def reduce(value1: (String, Int), value2: (String, Int)): (String, Int) = {

          (value1._1, value2._2 + value2._2)
        }
      }, new WindowFunction[(String, Int), (String, Int), String, TimeWindow] {
        override def apply(key: String, window: TimeWindow, input: Iterable[(String, Int)], out: Collector[(String, Int)]): Unit = {

          println(s"start-end: ${window.getStart} - ${window.getEnd}")
          for (elem <- input) {
            out.collect(elem)
          }

        }
      }

      ).print()


    env.execute()
  }

}
